<?php

namespace AsyncQueue;

use Kiri\Cache\Redis;
use Kiri\Crontab\Producer;
use Kiri;
use Kiri\Server\Contract\OnProcessInterface;
use Swoole\Coroutine;
use Swoole\Process;


/**
 *
 */
class AsyncQueueProcess implements OnProcessInterface
{


	/**
	 * @param Process $process
	 * @return string
	 */
	public function getProcessName(Process $process): string
	{
		// TODO: Implement getProcessName() method.
		return 'async queue process';
	}


	/**
	 * @param Process $process
	 */
	public function process(Process $process): void
	{
		// TODO: Implement onHandler() method.

		while (true) {
			$this->execute(static::loadCarobTask());
			Coroutine::sleep(0.03);
		}
	}


	/**
	 * @param array $tasks
	 */
	private function execute(array $tasks)
	{
		foreach ($tasks as $task) {
			$task = unserialize($task);
			if (!($task instanceof ConsumerInterface)) {
				continue;
			}
			go(static function () use ($task) {
				$task->onHandler();
			});
		}
	}


	/**
	 * @return array
	 * @throws \Exception
	 */
	public static function loadCarobTask(): array
	{
		$redis = Kiri::getDi()->get(Redis::class);
		$script = <<<SCRIPT
local _two = redis.call('zRangeByScore', KEYS[1], '0', ARGV[1])

if (table.getn(_two) > 0) then
	redis.call('ZREM', KEYS[1], unpack(_two))
end

return _two
SCRIPT;
		return $redis->eval($script, [Producer::CRONTAB_KEY, (string)time()], 1);
	}

	public function signListen(Process $process): void
	{
		// TODO: Implement signListen() method.
	}

	public function onProcessStop(): void
	{
		// TODO: Implement onProcessStop() method.
	}
}
